package vn.com.kms.storm.sample.topology;

import vn.com.kms.storm.sample.bolt.PrinterBolt;
import vn.com.kms.storm.sample.spout.TwitterSampleSpout;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;

/**
 * @see backtype.storm.topology.TopologyBuilder
 */
public class TwitterSampleTopology {

    public static void main(String[] args) throws Exception {
        String username = args[args.length - 2];
        String pwd = args[args.length - 1];
        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("twitter-spout", new TwitterSampleSpout(username, pwd));
        builder.setBolt("print", new PrinterBolt()).shuffleGrouping("twitter-spout");

        Config conf = new Config();

        if (args != null && args.length == 3) {
            conf.setNumWorkers(3);

            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else if (args.length == 2) {
            conf.setMaxTaskParallelism(3);

            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("testTwitter", conf, builder.createTopology());

            Utils.sleep(10000);
            // cluster.killTopology("testTwitter");
            cluster.shutdown();
        } else {
            System.out.println("Usage: TwitterSampleTopology [topologyName] username pwd");
            System.out.println("(if topology name is omitted, run in local mode)");
        }
    }

}
